package org.apache.flink.web.domain;

import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.types.logical.*;

import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;

public class SqlResult {
    /**
     * sessionId
     */
    private String sessionId;
    /**
     * 最终状态
     */
    private String state = "SUCCESS";
    /**
     * 提示消息
     */
    private String message = "";
    /**
     * 执行的Job列表
     */
    private List<JobResult> jobs = new ArrayList<>();

    public SqlResult() {
        this.sessionId = "default";
    }

    public SqlResult(String sessionId) {
        this.sessionId = sessionId;
    }

    public String getSessionId() {
        return sessionId;
    }

    public void setSessionId(String sessionId) {
        this.sessionId = sessionId;
    }

    public String getState() {
        return state;
    }

    public void setState(String state) {
        this.state = state;
    }

    public String getMessage() {
        return message;
    }

    public void setMessage(String message) {
        this.message = message;
    }

    public void failed(Throwable e) {
        this.setState("FAILED");
        while (e.getCause() != null && e.getCause() != e && e.getCause().getMessage() != null && !e.getCause().getMessage().isEmpty()) {
            e = e.getCause();
        }
        this.setMessage(e.getMessage());
    }

    public List<JobResult> getJobs() {
        return jobs;
    }

    public void setJobs(List<JobResult> jobs) {
        this.jobs = jobs;
    }

    public void addJob(String jobId, ResultColumn[] columns, List<String[]> datas) {
        this.addJob(newJobResult(jobId, columns, datas));
    }

    public void addJob(JobResult job) {
        if (job != null) {
            this.jobs.add(job);
        }
    }

    public static JobResult newJobResult(String jobId, ResultColumn[] columns, List<String[]> datas) {
        JobResult job = new JobResult();
        job.setJobId(jobId);
        job.setColumns(columns);
        job.setDatas(datas);
        return job;
    }

    public static JobResult newJobResult(String jobId) {
        return newJobResult(jobId, null, null);
    }
    public static class JobResult {
        /**
         * Job唯一Id
         */
        private String jobId;
        /**
         * 返回结果列
         */
        private ResultColumn[] columns;
        /**
         * 返回结果值
         */
        private List<String[]> datas;

        public String getJobId() {
            return jobId;
        }

        public void setJobId(String jobId) {
            this.jobId = jobId;
        }

        public ResultColumn[] getColumns() {
            return columns;
        }

        public void setColumns(ResultColumn[] columns) {
            this.columns = columns;
        }

        public List<String[]> getDatas() {
            return datas;
        }

        public void setDatas(List<String[]> datas) {
            this.datas = datas;
        }
    }

    public static class ResultColumn {
        /**
         * 名称
         */
        private String name;
        /**
         * 类型
         */
        private String type;
        /**
         * 类型
         */
        private String typeRoot;
        /**
         * 可以为空
         */
        private boolean isNullable = true;
        /**
         * 可以为空
         */
        private int precision = 0;
        /**
         * 可以为空
         */
        private int scale = 0;

        public ResultColumn() {
        }

        public ResultColumn(Column column) {
            this.name = column.getName();
            LogicalType logicalType = column.getDataType().getLogicalType();
            this.type = logicalType.toString();
            if(this.type.endsWith(" NOT NULL")) {
                this.type = this.type.substring(0, this.type.length() - " NOT NULL".length());
            }
            this.typeRoot = logicalType.getTypeRoot().name();
            this.isNullable = logicalType.isNullable();

            for (Field field : logicalType.getClass().getFields()) {
                try {
                    if(field.getName().equals("PRECISION")) {
                        this.precision = field.getInt(logicalType);
                    }
                    switch (field.getName()) {
                        case "PRECISION":
                        case "LENGTH":
                            this.precision = field.getInt(logicalType);
                            break;
                        case "SCALE":
                            this.scale = field.getInt(logicalType);
                            break;
                    }
                } catch (IllegalAccessException e) {
                    e.printStackTrace();
                }
            }

            Method[] declaredMethods = logicalType.getClass().getDeclaredMethods();
            for (Method declaredMethod : declaredMethods) {
                try {
                    switch (declaredMethod.getName()) {
                        case "getPrecision":
                        case "getLength":
                            this.precision = (int) declaredMethod.invoke(logicalType);
                            break;
                        case "getScale":
                            this.scale = (int) declaredMethod.invoke(logicalType);
                            break;
                    }
                } catch (IllegalAccessException | InvocationTargetException e) {
                    e.printStackTrace();
                }
            }
        }

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

        public String getType() {
            return type;
        }

        public void setType(String type) {
            this.type = type;
        }

        public String getTypeRoot() {
            return typeRoot;
        }

        public void setTypeRoot(String typeRoot) {
            this.typeRoot = typeRoot;
        }

        public int getScale() {
            return scale;
        }

        public void setScale(int scale) {
            this.scale = scale;
        }

        public boolean isNullable() {
            return isNullable;
        }

        public void setNullable(boolean nullable) {
            isNullable = nullable;
        }

        public int getPrecision() {
            return precision;
        }

        public void setPrecision(int precision) {
            this.precision = precision;
        }
    }

}
